package org.apache.spark.examples

import org.apache.spark.{SparkConf, SparkContext}

object SimpleGroupCalculation {
  def main(args: Array[String]): Unit = {
    // 创建SparkConf和SparkContext
    val conf = new SparkConf()
      .setAppName("Simple Group Calculation")
      .setMaster("local[*]")
    val sc = new SparkContext(conf)

    // 1. 创建原始数据：(姓名, 分数)
    val scoresData = List(
      ("Alice", 85),
      ("Bob", 92),
      ("Charlie", 78),
      ("David", 95),
      ("Alice", 88),
      ("Bob", 90),
      ("Charlie", 82)
    )

    // 2. 并行化数据为RDD
    // stage 0
    val scoresRDD = sc.parallelize(scoresData, 2)

    // stage 1
    // 3. 按姓名分组并计算平均分
    // groupByKey是一个宽依赖的转换操作
    // 它会触发shuffle，因此会创建一个新的Stage（Stage 1）
    // shuffle的原因：需要将相同key的数据重新分配到同一个分区
    val averageScores = scoresRDD
      .groupByKey() // 触发Stage 1的开始
      .mapValues(scores => {
        // mapValues是一个窄依赖的转换
        // 不会创建新的Stage，仍然在Stage 1中
        val total = scores.sum
        val count = scores.size
        total.toDouble / count
      })

    println("每个学生的平均分：")
    averageScores.collect().foreach { case (name, avgScore) =>
      println(f"$name: $avgScore%.2f")
    }

    // Stage划分的主要触发点：
    // - 宽依赖操作（如groupByKey）
    // - 行动算子（如collect）

    // 暂停以便查看Spark UI
    Thread.sleep(300000)

    sc.stop()
  }
} 